package com.i360day.redis.listener;

import com.alibaba.fastjson.JSON;
import com.i360day.redis.constant.RedisContextHolder;
import com.i360day.redis.listener.channel.RedisMessageListenerChannel;
import com.i360day.redis.listener.vo.RedisResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

//import com.i360day.redis.listener.dto.RedisResult;

/**
 * @description: redis消息总线
 * @author: liju.z
 * @create: 2019-09-05 19:15
 **/
public abstract class RedisMessageListenerAdapterSupport<T> extends MessageListenerAdapter {
    private Logger log = LoggerFactory.getLogger(RedisMessageListenerAdapterSupport.class);

    protected final Type type;

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description bean前置处理 <p>
     *
     * <p> @Date  16:28 <p>
     *
     * <p> @Param [] <p>
     *
     * <p> @return [] <p>
     **/
    public void afterPropertiesSet() {
        setDefaultListenerMethod("redisReturnMessage");
        super.afterPropertiesSet();
    }

    protected RedisMessageListenerAdapterSupport() {
        Type superClass = this.getClass().getGenericSuperclass();
        if (!(superClass instanceof ParameterizedType)) {
            logger.warn(String.format("Warn: %s's superclass not ParameterizedType", new Object[]{this.getClass().getSimpleName()}));
            this.type = Object.class;
            return;
        }
        Type[] params = ((ParameterizedType) superClass).getActualTypeArguments();
        if (params.length == 0) {
            logger.warn(String.format("Warn: Index: %s, Size of %s's Parameterized Type: %s .", new Object[]{0, this.getClass().getSimpleName(), Integer.valueOf(params.length)}));
            this.type = Object.class;
            return;
        }
        this.type = params[0];
    }

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description 自定义消息处理, 此类如果没代理了则无法识别 <p>
     *
     * <p> @Date  16:27 <p>
     *
     * <p> @Param [message] <p>
     *
     * <p> @return [message] <p>
     **/
    public abstract void handleMessage(T message);

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description 消息监听后，解析调用 <p>
     *
     * <p> @Date  14:47 <p>
     *
     * <p> @Param [message] <p>
     *
     * <p> @return [message] <p>
     **/

    public final void redisReturnMessage(String message) {
        //返回结果
        RedisResult redisResult = new RedisResult();
        //消息包
        RedisMessageListenerChannel messageListenerEvent = JSON.parseObject(message, RedisMessageListenerChannel.class);
        if (messageListenerEvent == null) {
            log.warn("通道转换器获取为空...obj={}, message={}", messageListenerEvent, message);
            redisResult.setSuccess(false);
            redisResult.setErrorMsg("通道转换器获取为空");
            callback(redisResult, messageListenerEvent);
            return;
        } else {
            String contentJson = JSON.toJSONString(messageListenerEvent.getMessage());
            String queueKey = RedisContextHolder.getQueueLeftKey(messageListenerEvent.getSourceClazzName(), messageListenerEvent.getQueueId());
            //只有是队列的时候才使用单线程执行
            if (messageListenerEvent.isQueue()) {
                //TODO 这里是队列单线程获取，肯定不会有并发现象
                Object leftPopObj = RedisContextHolder.leftPop(queueKey);
                if (leftPopObj == null) {
                    log.warn("列队已被取完...{}", message);
                    return;
                }
                RedisContextHolder.delete(queueKey);
            }
            try {
                T typeObj = JSON.parseObject(contentJson, type);
                //使用过滤器
                if (beforeFilter(typeObj)) {
                    this.handleMessage(typeObj);
                    redisResult.setSuccess(true);
                    callback(redisResult, messageListenerEvent);
                } else {
                    redisResult.setSuccess(false);
                    redisResult.setErrorMsg("条件不匹配");
                    callback(redisResult, messageListenerEvent);
                }
            } catch (Throwable ex) {
                redisResult.setSuccess(false);
                redisResult.setErrorMsg(ex.getMessage());
                callback(redisResult, messageListenerEvent);
                throw ex;
            }
        }
    }

    /**
     * <p> @Author liju.z <p>
     *
     * <p> @Description 过滤器 <p>
     *
     * <p> @Date  10:32 <p>
     *
     * <p> @Param [message] <p>
     *
     * <p> @return [message] <p>
     **/
    protected boolean beforeFilter(T message) {
        return true;
    }

    private void callback(RedisResult redisResult, RedisMessageListenerChannel messageListenerEvent) {
        if (messageListenerEvent.isSynch()) {
            String queueKey = RedisContextHolder.getQueueRightKey(messageListenerEvent.getSourceClazzName(), messageListenerEvent.getQueueId());
            RedisContextHolder.rightPush(queueKey, redisResult);
            RedisContextHolder.setExpire(queueKey, 60 * 10);
        }
    }

    @Override
    public String toString() {
        return "RedisMessageListenerAdapterSupport{" +
                "type=" + type +
                '}';
    }
}
